package org.util.mqttclient;

import java.net.URISyntaxException;
import java.util.LinkedList;

import org.fusesource.hawtbuf.AsciiBuffer;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;

public class TopicPublisher {
	
	private MQTT mqtt;
	
	public TopicPublisher(String host,int port,String username,String password) throws URISyntaxException{
		mqtt = new MQTT();
		mqtt.setHost(host, port);
		mqtt.setUserName(username);
		mqtt.setPassword(password);
	}
	
	public void publish(String topicName,String body) throws Exception {
		String destination =  topicName;
		System.out.println("push topic: "+destination);
		Buffer msg = new AsciiBuffer(body);

		FutureConnection connection = mqtt.futureConnection();
		connection.connect().await();

		final LinkedList<Future<Void>> queue = new LinkedList<Future<Void>>();
		UTF8Buffer topic = new UTF8Buffer(destination);

		// Send the publish without waiting for it to complete. This allows us
		// to send multiple message without blocking..
		queue.add(connection.publish(topic, msg, QoS.AT_LEAST_ONCE, false));

		// Eventually we start waiting for old publish futures to complete
		// so that we don't create a large in memory buffer of outgoing
		// message.s
		queue.removeFirst().await();

		while (!queue.isEmpty()) {
			queue.removeFirst().await();
		}
		connection.disconnect().await();
	}

}
